package com.afdor.rws.event.driven.channel.rocketmq.producer.topic.transaction.impl;

import com.afdor.rws.event.driven.api.channel.producer.EventProducer;
import com.afdor.rws.event.driven.api.channel.producer.transaction.EventProducerTransactionItem;
import com.afdor.rws.event.driven.channel.rocketmq.producer.topic.RocketMqTransactionEventProducer;
import com.afdor.rws.event.driven.entities.Event;
import lombok.Getter;
import org.apache.rocketmq.client.producer.LocalTransactionState;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.ConcurrentMap;

/**
 * 默认执行通道事物管理模式
 *
 * @author 悭梵
 * @date Created in 2018-12-17 16:32
 */
public class RocketMqTransactionItemImpl implements EventProducerTransactionItem {

    @Getter
    private final EventProducer eventProducer;

    @Getter
    private final String transactionId;

    @Getter
    private volatile List<Event> eventList;

    private final ConcurrentMap<String, RocketMqTransactionEventProducer.RocketMqTransactionItem> transactionEventMap;

    public RocketMqTransactionItemImpl(final String eventProducerTransactionId, final EventProducer eventProducer, final ConcurrentMap<String, RocketMqTransactionEventProducer.RocketMqTransactionItem> transactionEventMap) {
        this.eventProducer = eventProducer;
        this.transactionEventMap = transactionEventMap;
        this.transactionId = eventProducerTransactionId;
    }

    @Override
    public synchronized void addEvent(Event event) {
        if (this.eventList == null) {
            this.eventList = new ArrayList<>(10);
        }
        this.eventList.add(event);
        event.setStatus(Event.Status.JOIN_TRANSACTION);
        eventProducer.sendAsyncEvent(event);
    }

    @Override
    public synchronized void rollback() {
        Optional.ofNullable(getEventList())
                .orElse(Collections.emptyList())
                .forEach(event -> {
                    event.setStatus(Event.Status.ROLLBACK);
                    transactionEventMap.get(event.getId()).setLocalTransactionState(LocalTransactionState.ROLLBACK_MESSAGE);
                });
    }

    @Override
    public synchronized boolean commit() {
        Optional.ofNullable(getEventList())
                .orElse(Collections.emptyList())
                .forEach(event -> {
                    event.setSendType(Event.SendType.ASYNC_TRANSACTION);
                    event.setStatus(Event.Status.COMMITTED);
                    transactionEventMap.get(event.getId()).setLocalTransactionState(LocalTransactionState.COMMIT_MESSAGE);
                });
        return Boolean.TRUE;
    }

}
